refactor(rollout): engine-level sync/async contracts for async rollout - #4

Summary

Unifies the two async rollout paths behind engine-level contracts under unirl/rollout/engine/, replacing the over-layered async_runtime.py (scheduler + one-implementation dispatcher Protocol + callback types) and the agentic trainers' duplicated producer plumbing.

• engine/synchronous.py (was base.py): worker-side sync contracts. BaseSingleTurnRolloutEngine is renamed SyncRolloutEngine; chunked_engine_generate moves in from the package __init__.
• engine/asynchronous.py (new): the driver side — VersionedBuffer + InflightPool mechanisms, the AsyncRolloutEngine protocol (poll / drain_freshest / pop_evicted / quiesce, engine-owned weight_version), and two concretes: AsyncBatchRolloutEngine (AR/diffusion; launch-time version stamps) and AsyncAgenticRolloutEngine (facade over the agentic rank-0 coordinator; [0] unwraps, PendingGroups sibling assembly, completion-time stamps, quiesce folds abort + the mandatory follow-up poll).
• Handle gains a public non-blocking seam (launch_nowait → PendingHandleCall.ready()/wait()/result()), removing RayGenerationDispatcher's documented "update in lockstep" coupling to the private _execute_all/_rebind_tree; handle_fn is untouched.
• The four async trainers keep policy as explicit loops (launch ceiling, reap-vs-launch as statement order — diffusion polls before topping up, preserving the measured 150s→8s BAGEL constraint; quiesce before sync; tail carry/drop). The reap_before_launch flag, dead drained()/pending_roots()/_pending_carried, and trainer-side _weight_version counters are removed.
• engine/__init__.py becomes a lazy re-export shim so the driver-side module imports ray/torch-free, enabling the new CPU-only unit tests.

Net −13 lines with tests added; the runtime layer itself shrinks ~390 → ~450 lines covering both paths instead of one.

Related Issue

N/A (tracked in Linear as LIN-631).

Test Plan

• ruff check unirl/rollout unirl/trainer unirl/distributed/group/handle.py tests — passes.
• python3 -m compileall -q unirl/rollout unirl/trainer unirl/distributed/group/handle.py tests — passes.
• uvx pytest tests/ -q — 24 passed (new CPU-only tests pinning: complete-or-nothing reap retry without double-insert, first-error deferral, KeyboardInterrupt passthrough, freshest-first staleness eviction, launch-time vs completion-time version stamping, quiesce-poll-before-bump ordering, n-sibling group assembly).
• Not run (needs GPUs): end-to-end validation of examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml, the BAGEL bagel_vllmomni_async.yaml, and an ALFWorld partial/async recipe — recommended before merge; the diffusion reap-ordering and launch-ceiling invariants are only fully observable there.

Compatibility / Risk

• No recipe/config changes required; _target_ paths and knobs are unchanged (reap_before_launch was internal, never a recipe key).
• Import paths changed: unirl.rollout.async_runtime / deleted names (AsyncRolloutScheduler, RayGenerationDispatcher, VersionedGroupBuffer, GenerationDispatcher, InflightGeneration) are gone with no aliases; unirl.rollout.engine.base → unirl.rollout.engine.synchronous. All in-repo consumers are migrated; out-of-tree code importing these must update.
• Behavior-preserving by design; the riskiest spots are the per-trainer _next_step loops (reviewed line-by-line against the deleted next_step for both orderings) and the agentic quiesce() fold (test-pinned: groups completing during the quiesce carry the pre-sync version).
• _pending_carried removal is a verified dead-state deletion (only ever assigned []).

Reviewer Notes

• Suggested review order: handle.py seam (commit 1, additive) → engine/asynchronous.py mechanisms → protocol + engines → async_ar/async_diffusion loops (check reap ordering) → agentic_partial/agentic_async (check quiesce-before-bump and tail-policy counter divergence) → tests as the executable spec.
• AI-assisted (Claude Code); the full diff was reviewed and the listed checks run locally by the submitter. Checked for duplicate work: no overlapping open PRs.
• Follow-ups deliberately out of scope: trainer-level dedup of the still-duplicated _next_step/_build_tasks loop pairs, making handle_fn delegate to the seam halves, and the _connect_separate/two-slab-build duplication.

Checklist

• I reviewed the changed code and removed unrelated/generated artifacts.
• I updated tests, docs, and configs where needed, or explained why not.